package org.ghost.akka.client;

import akka.actor.AbstractActor;
import akka.actor.ActorRef;
import akka.event.Logging;
import akka.event.LoggingAdapter;
import akka.pattern.Patterns;
import akka.util.Timeout;
import org.ghost.akka.common.command.BaseCommand;
import org.ghost.akka.common.em.ResponseEnum;
import org.ghost.akka.common.model.JobResult;
import scala.concurrent.Await;
import scala.concurrent.Future;
import scala.concurrent.duration.Duration;

import java.util.concurrent.TimeUnit;

/**
 * 任务发送器
 */
public class TaskSendActor extends AbstractActor {
    private LoggingAdapter log = Logging.getLogger(getContext().system(), this);
    private Timeout timeout = new Timeout(Duration.create(5, TimeUnit.SECONDS));
    private ActorRef dispatcher;
    private AkkaClusterConfig config;

    @Override
    public Receive createReceive() {
        return receiveBuilder()
                .match(BaseCommand.class, command -> {
                    log.info("taskId=" + command.getJob().getTaskId());
                    Future<Object> future = Patterns.ask(dispatcher, command, timeout);
                    JobResult result = null;
                    try {
                        result = (JobResult) Await.result(future, timeout.duration());
                    } catch (Exception e) {
                        log.error("Message received from Server error-> {}", e);
                        result = new JobResult(ResponseEnum.TIME_OUT);
                    }
                    command.getJob().setResponse(result);
                    log.info("Message received from Server -> {}", result);
                    synchronized (command.getJob()) {
                        //notify
                        command.getJob().notify();
                    }
                }).build();
    }

    @SuppressWarnings("deprecation")
    @Override
    public void preStart() {
//		AkkaClusterConfig config = ContextUtils.getBean(AkkaClusterConfig.class, "clusterConfig");
        config = new AkkaClusterConfig();
        // Get a reference to the remote actor
        dispatcher = getContext().actorFor(config.buildRemoteUri());
    }
}
